package learn4ConfirmMessage

import com.rabbitmq.client.ConfirmCallback
import utils.RabbitMqUtils

import java.util.concurrent.ConcurrentSkipListMap

/**

 @author HSH

 @create 2022-01-07 10:14
  发布确认模式
  1.单个确认
  2.批量确认
  3.异步批量确认
 */
class ConfirmMessage {
    static MESSAGE_COUNT = 1000

    static void main(String[] args) {
        //1.单个确认 发布1000个单独确认消息，耗时2916ms
        publishMsgIndividually()
        //2.批量确认 发布1000个批量确认消息，耗时1448ms
        publishMsgBatch()
        //3.异步批量确认
        publishMsgAsync()

    }

    static publishMsgIndividually() {
        def startTime = System.currentTimeMillis()
        def channel = RabbitMqUtils.getChannel()
        String queueName = UUID.randomUUID().toString()
        /****
         * 生成一个队列
         * 1.队列名称
         * 2.队列里的消息是否持久化 默认消息存储在内存中
         * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
         * 4.是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除 true 自动删除
         * 5.其他参数
         *
         */
        channel.queueDeclare(queueName, true, false, false, null)
        //确认
        channel.confirmSelect()
        (0..MESSAGE_COUNT).each {
            channel.basicPublish("", queueName, null, it.toString().getBytes())
            if (channel.waitForConfirms()) {
//                println "${it}发送成功"
            }
        }
        def end = System.currentTimeMillis()
        println "发布${MESSAGE_COUNT}个单独确认消息，耗时${end - startTime}ms"

    }

    static publishMsgBatch() {
        def startTime = System.currentTimeMillis()
        def channel = RabbitMqUtils.getChannel()
        String queueName = UUID.randomUUID().toString()
        /****
         * 生成一个队列
         * 1.队列名称
         * 2.队列里的消息是否持久化 默认消息存储在内存中
         * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
         * 4.是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除 true 自动删除
         * 5.其他参数
         *
         */
        channel.queueDeclare(queueName, true, false, false, null)
        //确认
        channel.confirmSelect()
        int batchSize = 100
        (0..MESSAGE_COUNT).each {
            channel.basicPublish("", queueName, null, it.toString().getBytes())
            if (it > 0 && it % batchSize == 0) {
                if (channel.waitForConfirms()) {
//                    println "${it}发送成功"
                }
            }
        }
        def end = System.currentTimeMillis()
        println "发布${MESSAGE_COUNT}个批量确认消息，耗时${end - startTime}ms"
    }

    static publishMsgAsync() {
        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目 只要给到序列号
         * 3.支持并发访问
         */
        def concurrentSkipListMap = new ConcurrentSkipListMap()
        def startTime = System.currentTimeMillis()
        def channel = RabbitMqUtils.getChannel()
        String queueName = UUID.randomUUID().toString()
        /****
         * 生成一个队列
         * 1.队列名称
         * 2.队列里的消息是否持久化 默认消息存储在内存中
         * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
         * 4.是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除 true 自动删除
         * 5.其他参数
         *
         */
        channel.queueDeclare(queueName, true, false, false, null)
        //确认
        channel.confirmSelect()
        //准备消息的监听器 监听消息成功了那些失败了
        channel.addConfirmListener(new ConfirmCallback() {
            @Override
            void handle(long deliveryTag, boolean multiple) throws IOException {
                if(multiple){
                    def confirmedMap = concurrentSkipListMap.headMap(deliveryTag,true)
                    confirmedMap.clear()
                }else{
                    concurrentSkipListMap.remove(deliveryTag)
                }
//                println "线程id" + Thread.currentThread().getId()
                //消息确认成功回调
//                println "确认的消息${deliveryTag},是否为批量确认${multiple}"
            }
        }, new ConfirmCallback() {
            @Override
            void handle(long deliveryTag, boolean multiple) throws IOException {
                //消息确认失败回调
//                println "线程id" + Thread.currentThread().getId()
//                println "未确认的消息${deliveryTag},是否为批量确认${multiple}"
            }
        })

//        println "线程id" + Thread.currentThread().getId()
        (0..MESSAGE_COUNT).each {
            channel.basicPublish("", queueName, null, it.toString().getBytes())
            concurrentSkipListMap.put(channel.getNextPublishSeqNo(),it.toString())
        }


        def end = System.currentTimeMillis()
        println "发布${MESSAGE_COUNT}个异步确认消息，耗时${end - startTime}ms"
    }
}
